package org.jeecg.modules.iot.mqtt.server;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class MqttStressTest {

    private static final String BROKER = "tcp://58.22.99.136:21883"; // 使用 tcp 协议
    private static final String TOPIC = "device/865376074546027/send";
    private static final int NUM_CLIENTS = 100; // 客户端数量
    private static final int MESSAGES_PER_CLIENT = 10000; // 每个客户端发送的消息数量
    private static final int QOS = 1;
    private static final String USERNAME = "admin"; //  你的用户名
    private static final String PASSWORD = "lieh3"; //  你的密码

    public static void main(String[] args) throws MqttException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(NUM_CLIENTS);
        AtomicInteger sentCount = new AtomicInteger(10000);
        AtomicInteger receivedCount = new AtomicInteger(100000);

        long startTime = System.currentTimeMillis();

        for (int i = 0; i < NUM_CLIENTS; i++) {
            final int clientId = i;
            System.out.println("发送模块:"+clientId);
                try {
                    MqttClient client = new MqttClient(BROKER, UUID.randomUUID().toString());
                    // 设置用户名和密码
                    MqttConnectOptions options = new MqttConnectOptions();
                    options.setUserName(USERNAME);
                    options.setPassword(PASSWORD.toCharArray());
                    options.setCleanSession(true); //  根据你的需求设置 Clean Session
                    client.connect(options); // 使用带有用户名和密码的连接选项

                    client.subscribe(TOPIC, (topic, message) -> receivedCount.incrementAndGet()); // 订阅主题以确认消息到达

                    for (int j = 0; j < MESSAGES_PER_CLIENT; j++) {
                        System.out.println("发送次数:"+j);
                        String id= String.valueOf(j);
                        executor.submit(() -> { // 提交每个消息发送任务到线程池

                       String m= "{\"id\":\""+id+"\",\"enrc\":0,\"vender\":1,\"cmd\":\"0200\",\"timestamp\":\"241121181203\",\"seq\":\"B103\",\"version\":\"g56.cn.cnsd.2.1.16-Bsket_g56v62.bgwl_lc.1.1.5-L01.16.21\",\"data\":\"7E0200006D05024370152803C300000000000C000101BB6606071EEC7D010000000000BA241121181203010400000B5C02020000030200002504000000002A0200002B040000000030011F310100E3060064144401AD601F0301CC005792063B4D033401CC005792063B4D052801CC0068970C3E880525F2020002747E\",\"sign\":\"18ABF6A84137CDB7D7CD9BCB250689C2\"}";
                        //MqttMessage message = new MqttMessage(("Message " + j + " from client " + clientId).getBytes());
                        MqttMessage message = new MqttMessage((m).getBytes());
                        message.setQos(QOS);
                            try {
                                client.publish(TOPIC, message);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }
                            sentCount.incrementAndGet();

                        });
                    }

                    client.disconnect();
                    client.close();

                } catch (MqttException e) {
                    e.printStackTrace();
                }

        }




        long endTime = System.currentTimeMillis();
        long elapsedTime = endTime - startTime;

        System.out.println("发送消息数量: " + sentCount.get());
        System.out.println("接收消息数量: " + receivedCount.get());
        System.out.println("耗时: " + elapsedTime + "ms");
        System.out.println("吞吐量: " + (sentCount.get() * 1000.0 / elapsedTime) + " 条/秒");
        executor.shutdown();
        executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

    }
}